package com.im.service.message.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.im.codec.pack.message.MessageReadedPack;
import com.im.codec.pack.message.RecallMessageNotifyPack;
import com.im.common.constant.Constants;
import com.im.common.enums.ConversationTypeEnum;
import com.im.common.enums.DelFlagEnum;
import com.im.common.enums.exception.MessageErrorCode;
import com.im.common.enums.command.Command;
import com.im.common.enums.command.GroupEventCommand;
import com.im.common.enums.command.MessageCommand;
import com.im.common.model.ClientInfo;
import com.im.common.model.RestResponse;
import com.im.common.model.SyncReq;
import com.im.common.model.SyncResp;
import com.im.common.model.message.MessageReadedContent;
import com.im.common.model.message.MessageReciveAckContent;
import com.im.common.model.message.OfflineMessageContent;
import com.im.common.model.message.RecallMessageContent;
import com.im.service.conversation.service.ConversationService;
import com.im.service.group.service.ImGroupMemberService;
import com.im.service.message.entity.ImMessageBodyEntity;
import com.im.service.message.mapper.ImMessageBodyMapper;
import com.im.service.message.service.MessageSyncService;
import com.im.service.utils.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 * 消息同步服务类
 */
@Slf4j
@Service
public class MessageSyncServiceImpl implements MessageSyncService {

    @Autowired
    private MessageProducer messageProducer;
    @Autowired
    private ConversationService conversationService;
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private ImMessageBodyMapper imMessageBodyMapper;
    @Autowired
    private RedisSeq redisSeq;
    @Autowired
    private SnowflakeIdWorker snowflakeIdWorker;
    @Autowired
    private ImGroupMemberService imGroupMemberService;
    @Autowired
    private GroupMessageProducer groupMessageProducer;


    /**
     * 接收消息确认，发送给用户
     *
     * @param messageReciveAckContent 消息接收确认内容
     */
    public void receiveMark(MessageReciveAckContent messageReciveAckContent) {
        // 调用消息生产者发送消息给用户
        messageProducer.sendToUser(
                messageReciveAckContent.getToId(), // 接收消息的用户ID
                MessageCommand.MSG_RECIVE_ACK, // 消息类型为接收确认
                messageReciveAckContent, // 发送的消息内容为接收确认内容
                messageReciveAckContent.getAppId() // 发送的消息所属的应用ID
        );
    }


    /**
     * 消息已读。更新会话的seq，通知在线的同步端发送指定command ，发送已读回执通知对方（消息发起方）我已读
     *
     * @param messageContent 消息已读内容
     */
    public void readMark(MessageReadedContent messageContent) {
        // 调用会话服务的 messageMarkRead 方法更新会话的 seq
        conversationService.messageMarkRead(messageContent);

        // 创建一个 MessageReadedPack 对象，将 messageContent 的属性复制到该对象中
        MessageReadedPack messageReadedPack = new MessageReadedPack();
        BeanUtils.copyProperties(messageContent, messageReadedPack);

        // 同步已读消息给消息发送方
        syncToSender(
                messageReadedPack, // 同步的消息内容为已读消息
                messageContent, // 原始消息内容
                MessageCommand.MSG_READED_NOTIFY // 同步的消息类型为已读通知
        );

        // 发送已读回执通知给消息接收方
        messageProducer.sendToUser(
                messageContent.getToId(), // 消息接收方的用户ID
                MessageCommand.MSG_READED_RECEIPT, // 消息类型为已读回执通知
                messageReadedPack, // 发送的消息内容为已读消息
                messageContent.getAppId() // 发送的
                // 消息所属的应用ID
        );
    }


    /**
     * 将消息读取状态同步到接收者的其他客户端
     *
     * @param pack    消息读取包
     * @param content 消息读取内容
     * @param command 命令
     */
    public void syncToSender(MessageReadedPack pack, MessageReadedContent content, Command command) {
        // 该方法会向所有与发送者不同的客户端发送消息，以通知它们该消息已被读取
        messageProducer.sendToUserExceptClient(pack.getFromId(), command, pack, content);
    }


    /**
     * 将群组消息标记为已读
     *
     * @param messageReaded 包含已读消息信息的对象
     */
    public void groupReadMark(MessageReadedContent messageReaded) {
        // 调用conversationService的messageMarkRead方法，将消息标记为已读
        conversationService.messageMarkRead(messageReaded);
        // 创建MessageReadedPack对象，并将messageReaded对象的属性复制到messageReadedPack对象中
        MessageReadedPack messageReadedPack = new MessageReadedPack();
        BeanUtils.copyProperties(messageReaded, messageReadedPack);
        // 调用syncToSender方法，将消息已读状态同步给接收方的其他客户端
        syncToSender(messageReadedPack, messageReaded, GroupEventCommand.MSG_GROUP_READED_NOTIFY);
        // 如果消息发送方和接收方不是同一个人，则调用messageProducer的sendToUser方法，将已读消息发送给发送方
        if (!messageReaded.getFromId().equals(messageReaded.getToId())) {
            List<ClientInfo> clientInfos = messageProducer.sendToUser(messageReadedPack.getToId(), GroupEventCommand.MSG_GROUP_READED_RECEIPT, messageReaded, messageReaded.getAppId());
            log.info("将已读消息发送给发送方[{}]的所有客户端，发送成功的客户端列表：{}", messageReadedPack.getToId(), JSON.toJSONString(clientInfos));
        }
    }


    /**
     * 同步离线消息
     *
     * @param req 请求参数，包括appId、operater、lastSequence和maxLimit
     * @return RestResponse对象，包括SyncResp对象
     */
    public SyncResp syncOfflineMessage(SyncReq req) {
        //创建SyncResp对象
        SyncResp<OfflineMessageContent> resp = new SyncResp<>();
        //拼接Redis中离线消息的key
        String key = req.getAppId() + ":" + Constants.RedisConstants.OfflineMessage + ":" + req.getOperator();
        //获取最大的seq
        Long maxSeq = 0L;
        ZSetOperations zSetOperations = redisTemplate.opsForZSet();
        Set set = zSetOperations.reverseRangeWithScores(key, 0, 0);
        if (!CollectionUtils.isEmpty(set)) {
            //获取最大的seq
            List list = new ArrayList(set);
            DefaultTypedTuple o = (DefaultTypedTuple) list.get(0);
            maxSeq = o.getScore().longValue();
        }

        //创建离线消息列表
        List<OfflineMessageContent> respList = new ArrayList<>();
        //设置最大的seq
        resp.setMaxSequence(maxSeq);

        //查询Redis中指定范围内的离线消息
        Set<ZSetOperations.TypedTuple> querySet = zSetOperations.rangeByScoreWithScores(key,
                req.getLastSequence(), maxSeq, 0, req.getMaxLimit());
        for (ZSetOperations.TypedTuple<String> typedTuple : querySet) {
            //解析离线消息
            String value = typedTuple.getValue();
            OfflineMessageContent offlineMessageContent = JSONObject.parseObject(value, OfflineMessageContent.class);
            respList.add(offlineMessageContent);
        }
        resp.setDataList(respList);

        //判断是否查询到了离线消息
        if (!CollectionUtils.isEmpty(respList)) {
            //获取最后一条离线消息
            OfflineMessageContent offlineMessageContent = respList.get(respList.size() - 1);
            //判断是否查询完成
            resp.setCompleted(maxSeq <= offlineMessageContent.getMessageKey());
        }

        //返回RestResponse对象
        return resp;
    }


    /**
     * 处理消息撤回操作，具体步骤如下:
     * <p>
     * 1.修改历史消息的状态
     * 2.修改离线消息的状态
     * 3.发送确认消息给发送方
     * 4.发送同步消息给同步端
     * 5.分发撤回通知给消息接收方
     * </p>
     *
     * @param content 撤回消息内容
     */
    public void recallMessage(RecallMessageContent content) {
        Long messageTime = content.getMessageTime();
        Long now = System.currentTimeMillis();

        RecallMessageNotifyPack pack = new RecallMessageNotifyPack();
        BeanUtils.copyProperties(content, pack);

        // 判断消息是否超时（超过2分钟不能撤回）
        if (120000L < now - messageTime) {
            // 返回错误信息，通知撤回失败
            recallAck(pack, RestResponse.error(MessageErrorCode.MESSAGE_RECALL_TIME_OUT), content);
            return;
        }

        // 查询消息体
        QueryWrapper<ImMessageBodyEntity> query = new QueryWrapper<>();
        query.eq("app_id", content.getAppId());
        query.eq("message_key", content.getMessageKey());
        ImMessageBodyEntity body = imMessageBodyMapper.selectOne(query);

        if (body == null) {
            // 返回错误信息，通知撤回失败（消息不存在）
            recallAck(pack, RestResponse.error(MessageErrorCode.MESSAGEBODY_IS_NOT_EXIST), content);
            return;
        }

        if (body.getDelFlag() == DelFlagEnum.DELETE.getCode()) {
            // 返回错误信息，通知撤回失败（消息已被撤回）
            recallAck(pack, RestResponse.error(MessageErrorCode.MESSAGE_IS_RECALLED), content);
            return;
        }

        // 标记消息为已撤回
        body.setDelFlag(DelFlagEnum.DELETE.getCode());
        imMessageBodyMapper.update(body, query);

        if (content.getConversationType() == ConversationTypeEnum.P2P.getCode()) {

            // 找到fromId的队列
            String fromKey = content.getAppId() + ":" + Constants.RedisConstants.OfflineMessage + ":" + content.getFromId();
            // 找到toId的队列
            String toKey = content.getAppId() + ":" + Constants.RedisConstants.OfflineMessage + ":" + content.getToId();

            // 构造离线消息
            OfflineMessageContent offlineMessageContent = new OfflineMessageContent();
            BeanUtils.copyProperties(content, offlineMessageContent);
            offlineMessageContent.setDelFlag(DelFlagEnum.DELETE.getCode());
            offlineMessageContent.setMessageKey(content.getMessageKey());
            offlineMessageContent.setConversationType(ConversationTypeEnum.P2P.getCode());
            offlineMessageContent.setConversationId(conversationService.convertConversationId(offlineMessageContent.getConversationType(), content.getFromId(), content.getToId()));
            offlineMessageContent.setMessageBody(body.getMessageBody());

            // 生成消息序列号
            long seq = redisSeq.doGetSeq(content.getAppId() + ":" + Constants.SeqConstants.Message + ":" + ConversationIdGenerate.generateP2PId(content.getFromId(), content.getToId()));
            offlineMessageContent.setMessageSequence(seq);

            // 将离线消息存储到Redis中
            long messageKey = SnowflakeIdWorker.nextId();
            redisTemplate.opsForZSet().add(fromKey, JSONObject.toJSONString(offlineMessageContent), messageKey);
            redisTemplate.opsForZSet().add(toKey, JSONObject.toJSONString(offlineMessageContent), messageKey);

            // ack
            // 返回成功信息，通知撤回成功
            recallAck(pack, RestResponse.success(), content);
            // 分发给同步端
            messageProducer.sendToUserExceptClient(content.getFromId(), MessageCommand.MSG_RECALL_NOTIFY, pack, content);
            // 分发给对方
            messageProducer.sendToUser(content.getToId(), MessageCommand.MSG_RECALL_NOTIFY, pack, content.getAppId());
        } else {
            // 获取群组成员列表
            List<String> groupMemberId = imGroupMemberService.getGroupMemberId(content.getToId(), content.getAppId());
            // 生成消息序列号
            long seq = redisSeq.doGetSeq(content.getAppId() + ":" + Constants.SeqConstants.Message + ":" + ConversationIdGenerate.generateP2PId(content.getFromId(), content.getToId()));
            //ack
            // 返回成功信息，通知撤回成功
            recallAck(pack, RestResponse.success(), content);
            // 发送给同步端
            messageProducer.sendToUserExceptClient(content.getFromId(), MessageCommand.MSG_RECALL_NOTIFY, pack, content);
            for (String memberId : groupMemberId) {
                String toKey = content.getAppId() + ":" + Constants.SeqConstants.Message + ":" + memberId;
                OfflineMessageContent offlineMessageContent = new OfflineMessageContent();
                offlineMessageContent.setDelFlag(DelFlagEnum.DELETE.getCode());
                BeanUtils.copyProperties(content, offlineMessageContent);
                offlineMessageContent.setConversationType(ConversationTypeEnum.GROUP.getCode());
                offlineMessageContent.setConversationId(conversationService.convertConversationId(offlineMessageContent.getConversationType(), content.getFromId(), content.getToId()));
                offlineMessageContent.setMessageBody(body.getMessageBody());
                offlineMessageContent.setMessageSequence(seq);
                redisTemplate.opsForZSet().add(toKey, JSONObject.toJSONString(offlineMessageContent), seq);
                groupMessageProducer.producer(content.getFromId(), MessageCommand.MSG_RECALL_NOTIFY, pack, content);
            }
        }

    }

    /**
     * 处理消息撤回的确认操作。
     *
     * @param recallPack 撤回通知包含的信息
     * @param success    撤回操作成功的响应对象
     * @param clientInfo 客户端信息
     */
    public void recallAck(RecallMessageNotifyPack recallPack, RestResponse<Object> success, ClientInfo clientInfo) {
        // 将成功的响应对象赋值给 wrappedResp
        RestResponse<Object> wrappedResp = success;

        // 将消息发送给指定的用户，使用 MessageCommand.MSG_RECALL_ACK 命令，并携带 wrappedResp 和 clientInfo
        messageProducer.sendToUser(
                recallPack.getFromId(),  // 消息发送者的用户ID
                MessageCommand.MSG_RECALL_ACK,  // 使用的命令类型为 MSG_RECALL_ACK
                wrappedResp,  // 携带的响应数据对象
                clientInfo  // 携带的客户端信息对象
        );
    }


}
